//
//  stats_mcast_client.cpp
// ~~~~~~~~~~~~~~~~~~~~~
//
// Copyright (c) 2013-2014 Razvan Coca (razvan dot coca at  gmail dot com)
// This is mostly generated code. Generator is distributed under GPL. Edit it to suit your needs.
// 


#include "components/defs/handler_types.hpp"
#include "stats_mcast_client.hpp"

void stats_mcast_client::thread_func()
{
    while(true)
    {
        try{
            std::size_t qlen=ios_.run();
            LOG(ERROR)<<"service queue len:"<<qlen;
            if(!qlen)break;
        }
        catch(shutdown_exception const& e)
        {
            LOG(ERROR)<<"service thread error:"<<e.what();
            break;
        }
        catch(std::exception  const& e)
        {
            LOG(ERROR)<<"service thread error:"<<e.what();
        }
    }
}
stats_mcast_client::stats_mcast_client(ba::io_service& ios, udp_endpoint_t& local_if, udp_endpoint_t& mgroup, int ttl,
                                       crypto_buf_t& iv, crypto_buf_t& key,
                                       bt::millisec flight_base, bt::millisec flight_increment, int nitems,
                                       const std::string &cluster, const std::string &node):
    ios_(ios),
    flight_base_(flight_base),flight_increment_(flight_increment),nitems_(nitems),timer_(ios),
    worker_is_done_(false),
    work_(boost::make_shared<boost::asio::io_service::work>(ios)),
    thread_(boost::bind(&stats_mcast_client::thread_func,this)),
    processor_fsm_(cluster,node)
{
    name_=std::string(__FUNCTION__);
    mcast_client_process_=components::service::mcast_client_process::create(ios_,local_if,mgroup,ttl,iv,key,message_processor_);
    message_processor_.add(boost::bind(&components::protocol::stats_client_processor::process_request_set,boost::addressof(processor_fsm_),_1,_2,_3)).
            add(boost::bind(&components::protocol::stats_client_processor::process_request_get,boost::addressof(processor_fsm_),_1,_2,_3)).
            add(boost::bind(&components::protocol::stats_client_processor::process_timeout,boost::addressof(processor_fsm_),_1,_2,_3)).
            add(boost::bind(&components::protocol::stats_client_processor::process_end,boost::addressof(processor_fsm_),_1,_2,_3)).
            add(boost::bind(&components::protocol::stats_client_processor::process_reply,boost::addressof(processor_fsm_),_1,_2,_3));

}

void stats_mcast_client::handle_service_done (bs::error_code const& ec,const crypto_error& eec)
{
    LOG(INFO)<<name_<<"::"<<__FUNCTION__<<"["<<ec<<"]"<<"["<<eec.what()<<"]";
    //if(!ec)
    {
        boost::mutex::scoped_lock Lock(mutex_);
        if(ec)
        {
            worker_is_done_=true;
            condition_.notify_all();
            return;
        }
        if(!replies_.empty()&&replies_.back().type()==stats_message_t::end)
        {
            worker_is_done_=true;
            condition_.notify_all();
        }
    }
}

bool stats_mcast_client::query(stats_message_t & message)
{
    {
        boost::mutex::scoped_lock Lock(mutex_);
        worker_is_done_=false;
        replies_.clear();
    }
    phdl::system_crypto_handler_t
            handler=boost::bind(&stats_mcast_client::handle_service_done,this,ba::placeholders::error,_2);
    bool ret=mcast_client_process_->exec(flight_base_,flight_increment_,nitems_,message,boost::ref(replies_),handler);
    work_.reset();
    return ret;

}
std::vector<stats_message_t> stats_mcast_client::get_results()
{
    boost::mutex::scoped_lock Lock(mutex_);
    while (!worker_is_done_)condition_.wait(Lock);
    std::vector<stats_message_t> result(replies_);
    return result;
}
